

package com.hazelcast.internal.server.tcp;

import com.hazelcast.config.AdvancedNetworkConfig;
import com.hazelcast.config.Config;
import com.hazelcast.config.EndpointConfig;
import com.hazelcast.config.SSLConfig;
import com.hazelcast.instance.BuildInfoProvider;
import com.hazelcast.instance.EndpointQualifier;
import com.hazelcast.internal.networking.ChannelInitializer;
import com.hazelcast.internal.nio.ascii.TextChannelInitializer;
import com.hazelcast.internal.server.ServerContext;
import com.hazelcast.logging.ILogger;
import com.hazelcast.logging.Logger;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Function;

public class ChannelInitializerFunction implements Function<EndpointQualifier, ChannelInitializer> {
    protected static final ILogger LOGGER = Logger.getLogger(ChannelInitializerFunction.class);

    protected final ServerContext serverContext;
    private final ChannelInitializer uniChannelInitializer;
    private final Config config;
    private volatile Map<EndpointQualifier, ChannelInitializer> initializerMap;

    public ChannelInitializerFunction(ServerContext serverContext, Config config) {
        checkSslConfigAvailability(config);
        this.serverContext = serverContext;
        this.uniChannelInitializer = new UnifiedChannelInitializer(serverContext);
        this.config = config;
    }

    @Override
    public ChannelInitializer apply(EndpointQualifier qualifier) {
        return initializerMap.isEmpty() ? provideUnifiedChannelInitializer() : initializerMap.get(qualifier);
    }

    public void init() {
        AdvancedNetworkConfig advancedNetworkConfig = config.getAdvancedNetworkConfig();
        if (!advancedNetworkConfig.isEnabled() || advancedNetworkConfig.getEndpointConfigs().isEmpty()) {
            initializerMap = Collections.emptyMap();
            return;
        }

        Map<EndpointQualifier, ChannelInitializer> map = new HashMap<>();
        for (EndpointConfig endpointConfig : advancedNetworkConfig.getEndpointConfigs().values()) {
            checkSslConfigAvailability(endpointConfig.getSSLConfig());

            logSSLConfig(endpointConfig);

            switch (endpointConfig.getProtocolType()) {
            case MEMBER:
                map.put(EndpointQualifier.MEMBER, provideMemberChannelInitializer(endpointConfig));
                break;
            case CLIENT:
                map.put(EndpointQualifier.CLIENT, provideClientChannelInitializer(endpointConfig));
                break;
            case REST:
                map.put(EndpointQualifier.REST, provideTextChannelInitializer(endpointConfig, true));
                break;
            case MEMCACHE:
                map.put(EndpointQualifier.MEMCACHE, provideTextChannelInitializer(endpointConfig, false));
                break;
            case WAN:
                map.put(endpointConfig.getQualifier(), provideMemberChannelInitializer(endpointConfig));
                break;
            default:
                throw new IllegalStateException("Cannot build channel initializer for protocol type " + endpointConfig.getProtocolType());
            }
        }

        initializerMap = map;
    }

    private void logSSLConfig(EndpointConfig endpointConfig) {
        SSLConfig sslConfig = endpointConfig.getSSLConfig();
        if (sslConfig != null && sslConfig.isEnabled()) {
            switch (endpointConfig.getProtocolType()) {
            case MEMBER -> LOGGER.info("SSL is enabled for member-to-member connections");
            case CLIENT -> LOGGER.info("SSL is enabled for client-to-member connections");
            case WAN -> LOGGER.info("SSL is enabled for WAN connections");
            case REST -> LOGGER.info("SSL is enabled for REST connections");
            case MEMCACHE -> LOGGER.info("SSL is enabled for MEMCACHE connections");
            default -> LOGGER.info("Unknown Protocol Type");
            }
        }
    }

    protected ChannelInitializer provideUnifiedChannelInitializer() {
        return uniChannelInitializer;
    }

    protected ChannelInitializer provideMemberChannelInitializer(EndpointConfig endpointConfig) {
        return new MemberChannelInitializer(serverContext, endpointConfig);
    }

    protected ChannelInitializer provideClientChannelInitializer(EndpointConfig endpointConfig) {
        return new ClientChannelInitializer(serverContext, endpointConfig);
    }

    protected ChannelInitializer provideTextChannelInitializer(EndpointConfig endpointConfig, boolean rest) {
        return new TextChannelInitializer(serverContext, endpointConfig, rest);
    }

    protected ChannelInitializer provideWanChannelInitializer(EndpointConfig endpointConfig) {
        throw new UnsupportedOperationException("TODO");
    }

    // check SSL config for unisocket member configuration
    private void checkSslConfigAvailability(Config config) {
        if (config.getAdvancedNetworkConfig().isEnabled()) {
            return;
        }
        SSLConfig sslConfig = config.getNetworkConfig().getSSLConfig();
        checkSslConfigAvailability(sslConfig);
    }

    // check given SSL config
    private void checkSslConfigAvailability(SSLConfig sslConfig) {
        if (sslConfig != null && sslConfig.isEnabled()) {
            if (!BuildInfoProvider.getBuildInfo().isEnterprise()) {
                throw new IllegalStateException("SSL/TLS requires Hazelcast Enterprise Edition");
            }
        }
    }
}
